package com.mco.biz;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.*;
import com.alibaba.otter.canal.protocol.Message;

import com.mco.utils.RedisUtils;
import redis.clients.jedis.Jedis;
import java.net.InetSocketAddress;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.TimeUnit;

public class RedisCanalClient {
	public static final Integer _60SECONDS = 60;
	public static final String  CANAL_IP_ADDR = "192.168.1.11";
	
	private static void redisInsert(List<Column> columns)
	{
		JSONObject jsonObject = new JSONObject();
		for (Column column : columns)
		{
			System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
			jsonObject.put(column.getName(),column.getValue());
		}
		if(columns.size() > 0)
		{
			try(Jedis jedis = RedisUtils.getJedis())
			{
				jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
			}catch (Exception e){
				e.printStackTrace();
			}
		}
	}
	
	
	private static void redisDelete(List<Column> columns)
	{
		JSONObject jsonObject = new JSONObject();
		for (Column column : columns)
		{
			jsonObject.put(column.getName(),column.getValue());
		}
		if(columns.size() > 0)
		{
			try(Jedis jedis = RedisUtils.getJedis())
			{
				jedis.del(columns.get(0).getValue());
			}catch (Exception e){
				e.printStackTrace();
			}
		}
	}
	
	private static void redisUpdate(List<Column> columns)
	{
		JSONObject jsonObject = new JSONObject();
		for (Column column : columns)
		{
			System.out.println(column.getName() + " : " + column.getValue() + "    update=" + column.getUpdated());
			jsonObject.put(column.getName(),column.getValue());
		}
		if(columns.size() > 0)
		{
			try(Jedis jedis = RedisUtils.getJedis())
			{
				jedis.set(columns.get(0).getValue(),jsonObject.toJSONString());
				System.out.println("---------update after: "+jedis.get(columns.get(0).getValue()));
			}catch (Exception e){
				e.printStackTrace();
			}
		}
	}
	
	public static void printEntry(List<Entry> entrys)
	{
		for (Entry entry : entrys) {
			if (entry.getEntryType() == EntryType.TRANSACTIONBEGIN || entry.getEntryType() == EntryType.TRANSACTIONEND) {
				continue;
			}
			
			RowChange rowChage = null;
			try {
				//获取变更的row数据
				rowChage = RowChange.parseFrom(entry.getStoreValue());
			} catch (Exception e) {
				throw new RuntimeException("ERROR ## parser of eromanga-event has an error,data:" + entry.toString(),e);
			}
			//获取变动类型
			EventType eventType = rowChage.getEventType();
			System.out.println(String.format("================&gt; binlog[%s:%s] , name[%s,%s] , eventType : %s",
					entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
					entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), eventType));
			
			for (RowData rowData : rowChage.getRowDatasList()) {
				if (eventType == EventType.INSERT) {
					redisInsert(rowData.getAfterColumnsList());
				} else if (eventType == EventType.DELETE) {
					redisDelete(rowData.getBeforeColumnsList());
				} else {//EventType.UPDATE
					redisUpdate(rowData.getAfterColumnsList());
				}
			}
		}
	}
	
	
	public static void main(String[] args)
	{
		System.out.println("---------O(∩_∩)O哈哈~ initCanal() main方法-----------");
		
		//=================================
		// 创建链接canal服务端
		CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(CANAL_IP_ADDR,
				11111), "example", "", "");  // 这里用户名和密码如果在这写了，会覆盖canal配置文件的账号密码，如果不填从配置文件中读
		int batchSize = 1000;
		//空闲空转计数器
		int emptyCount = 0;
		System.out.println("---------------------canal init OK，开始监听mysql变化------");
		try {
			connector.connect();
			//connector.subscribe(".*\\..*");
			connector.subscribe("test.t_user");   // 设置监听哪个表
			connector.rollback();
			int totalEmptyCount = 10 * _60SECONDS;
			while (emptyCount < totalEmptyCount) {
				System.out.println("我是canal，每秒一次正在监听:"+ UUID.randomUUID().toString());
				Message message = connector.getWithoutAck(batchSize); // 获取指定数量的数据
				long batchId = message.getId();
				int size = message.getEntries().size();
				if (batchId == -1 || size == 0) {
					emptyCount++;
					try { TimeUnit.SECONDS.sleep(1); } catch (InterruptedException e) { e.printStackTrace(); }
				} else {
					//计数器重新置零
					emptyCount = 0;
					printEntry(message.getEntries());
				}
				connector.ack(batchId); // 提交确认
				// connector.rollback(batchId); // 处理失败, 回滚数据
			}
			System.out.println("已经监听了"+totalEmptyCount+"秒，无任何消息，请重启重试......");
		} finally {
			connector.disconnect();
		}
	}
}
